package com.shujia.flink.core

import com.alibaba.fastjson.{JSON, JSONObject}
import org.apache.flink.api.common.eventtime.{SerializableTimestampAssigner, WatermarkStrategy}
import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.connector.kafka.source.KafkaSource
import org.apache.flink.connector.kafka.source.enumerator.initializer.OffsetsInitializer
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.api.scala.function.ProcessWindowFunction
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows
import org.apache.flink.streaming.api.windowing.time.Time
import org.apache.flink.streaming.api.windowing.windows.TimeWindow
import org.apache.flink.util.Collector

import java.time.Duration

object Demo10CarTime {
  def main(args: Array[String]): Unit = {
    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment

    env.setParallelism(1)

    /**
     * 读取数据
     *
     */

    val source: KafkaSource[String] = KafkaSource
      .builder[String]
      .setBootstrapServers("master:9092,node1:9092,node2:9092") //kafka集群broker列表
      .setTopics("cars") //指定topic
      .setGroupId("asdasdasd") //指定消费者组，一条数据在一个组内只被消费一次
      .setStartingOffsets(OffsetsInitializer.latest()) //读取数据的位置，earliest：读取所有的数据，latest：读取最新的数据
      .setValueOnlyDeserializer(new SimpleStringSchema()) //反序列的类
      .build

    //使用kafka source
    val carsDS: DataStream[String] = env.fromSource(source, WatermarkStrategy.noWatermarks(), "Kafka Source")

    /**
     * 解析数据
     *
     */

    val cardAndTimeDS: DataStream[(Long, Long)] = carsDS.map(line => {
      //将字符串转换成json对象
      val jsonObj: JSONObject = JSON.parseObject(line)
      //使用字段名获取字段值
      //卡口编号
      val card: Long = jsonObj.getLong("card")
      //事件时间，事件时间要求时毫秒级别
      val time: Long = jsonObj.getLong("time") * 1000
      (card, time)
    })

    /**
     * 增加水位线和时间字段
     *
     */
    val assDS: DataStream[(Long, Long)] = cardAndTimeDS.assignTimestampsAndWatermarks(
      WatermarkStrategy
        //设置水位线的生成策略，前移5秒
        .forBoundedOutOfOrderness(Duration.ofSeconds(5))
        //设置时间字段
        .withTimestampAssigner(new SerializableTimestampAssigner[(Long, Long)] {
          override def extractTimestamp(element: (Long, Long), recordTimestamp: Long): Long = {
            //时间字段
            element._2
          }
        })
    )


    /**
     * 按照卡口分组
     *
     */
    val kvDS: DataStream[(Long, Int)] = assDS.map(kv => (kv._1, 1))

    //按照卡口分组
    val keyBYDS: KeyedStream[(Long, Int), Long] = kvDS.keyBy(_._1)

    //开窗口
    val windowDS: WindowedStream[(Long, Int), Long, TimeWindow] = keyBYDS
      .window(SlidingEventTimeWindows.of(Time.minutes(15), Time.minutes(4)))

    val cardFlowDS: DataStream[(Long, Long, Long, Int)] = windowDS
      .process(new ProcessWindowFunction[(Long, Int), (Long, Long, Long, Int), Long, TimeWindow] {
        /**
         * 一个窗口执行一次
         *
         * @param key      ：卡口
         * @param context  ：上下文对象
         * @param elements ：窗口内所有的数据
         * @param out      ： 用于将数据发送到下游
         */
        override def process(key: Long,
                             context: Context,
                             elements: Iterable[(Long, Int)],
                             out: Collector[(Long, Long, Long, Int)]): Unit = {

          //车流量
          val flow: Int = elements.size

          //获取窗口的开始和结束时间
          val window: TimeWindow = context.window
          val stat: Long = window.getStart
          val end: Long = window.getEnd

          //将数据发送到下游
          out.collect((key, stat, end, flow))
        }
      })


    cardFlowDS.print()

    env.execute()

  }

}
